package com.atguigu.day09;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

// keyBy在Flink SQL中如何写
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Tuple2<String, String>> stream = env
                .fromElements(
                        Tuple2.of("Mary", "./home"),
                        Tuple2.of("Bob", "./cart"),
                        Tuple2.of("Mary", "./prod?id=1"),
                        Tuple2.of("Liz", "./home")
                );

        // 获取表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);

        // 将数据流转化成动态表
        Table table = streamTableEnvironment
                .fromDataStream(
                        stream,
                        $("f0").as("user"),
                        $("f1").as("url")
                );

        // 将动态表注册为临时视图
        streamTableEnvironment
                .createTemporaryView("clicks", table);

        // 在clicks上做sql查询
        Table result = streamTableEnvironment
                .sqlQuery("SELECT user, COUNT(user) as cnt " +
                        "FROM clicks GROUP BY user");

        // 将结果动态表result转换成数据流
        // 由于sql中存在聚合操作，所以需要使用toChangeLogStream方法
        DataStream<Row> rowDataStream = streamTableEnvironment
                .toChangelogStream(result);

        rowDataStream.print();

        env.execute();
    }
}
